package org.damocode.iot.network.udp;

import io.netty.buffer.Unpooled;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.damocode.iot.core.message.PayloadType;
import org.eclipse.californium.elements.*;
import rx.subjects.PublishSubject;

import java.net.InetSocketAddress;

/**
 * @Description:
 * @Author: zzg
 * @Date: 2022/7/18 11:30
 * @Version: 1.0.0
 */
@Slf4j
public class DefaultUdpSupport implements UdpSupport {

    @Getter
    private final String id;

    private Connector connector;

    private InetSocketAddress address;

    private PublishSubject<UdpMessage> processor = PublishSubject.create();

    @Getter
    @Setter
    private String lastError;

    @Getter
    @Setter
    private InetSocketAddress bindAddress;

    public DefaultUdpSupport(String id) {
        this.id = id;
    }

    @Override
    public void shutdown() {
        if (null != connector) {
            connector.destroy();
            connector = null;
        }
    }

    @Override
    public void publish(UdpMessage message) {
        InetSocketAddress socketAddress = message.getAddress();
        if (socketAddress == null) {
            socketAddress = address;
        }
        if (socketAddress == null) {
            processor.error(new IllegalArgumentException("address cannot be null"));
            return;
        }
        if (connector == null) {
            return;
        }
        byte[] bytes = (byte[]) PayloadType.BINARY.read(message.getPayload());

        connector.send(RawData.outbound(bytes, new UdpEndpointContext(socketAddress), new MessageCallback() {
            @Override
            public void onConnecting() {

            }

            @Override
            public void onDtlsRetransmission(int flight) {

            }

            @Override
            public void onContextEstablished(EndpointContext context) {

            }

            @Override
            public void onSent() {
            }

            @Override
            public void onError(Throwable error) {
                processor.error(error);
            }
        }, false));
    }

    @Override
    public PublishSubject<UdpMessage> subscribe() {
        return this.processor;
    }

    @Override
    public boolean isAlive() {
        return connector != null;
    }

    @SneakyThrows
    public void setConnector(Connector connector) {
        this.connector = connector;
        this.connector.setRawDataReceiver(raw -> {
            log.info("raw:" + raw);
            byte[] payload = raw.getBytes();
            if (processor.hasObservers()) {
                processor.onNext(new UdpMessage(Unpooled.wrappedBuffer(payload), raw.getInetSocketAddress()));
            }
        });
        this.connector.start();
    }

    public void setAddress(InetSocketAddress address) {
        this.address = address;
    }
}
